#include "io_mqtt.h"

#include <stdlib.h>
#include <stdio.h>
#include <iostream>
#include <memory>
#include <mosquittopp.h>

#include "producer_mqtt.h"
#include "consumer_mqtt.h"
#include "business_def.h"
#include "Log.h"
#include "py_mqtt.h"
#ifdef WIN32
#define usleep(x) Sleep(x)
#endif

IOMQTT::IOMQTT(void)
	: running(true)
	, logf(false)
{
	mosqpp::lib_init();
	{
		BusinessDef *ptr_bdef = BusinessDef::getInstance();
		pyfree::MqttConf conf_ = ptr_bdef->getMqttConf();
		ip_s = conf_.ip;
		port_i = conf_.port;
		user_s = conf_.user;
		password_s = conf_.password;
		char buf[64] = { 0 };
		sprintf(buf, "pyfree_%d", ptr_bdef->getAreaInfo().areaId);
		mqttcpp = new PyMQTTIO((const char*)buf);
		mqttcpp->username_pw_set(user_s.c_str(), password_s.c_str());
		consumer = new ConsumerMQTT(mqttcpp);
		consumer->start();
		producer = new ProducerMQTT(mqttcpp);
		producer->start();
	}
}

IOMQTT::~IOMQTT(void)
{
	{
		running = false;
		if (NULL != producer) {
			delete producer;
			producer = NULL;
		}
		if (NULL != consumer) {
			delete consumer;
			consumer = NULL;
		}
		disconnect();
		cleanup();
	}
	mosqpp::lib_cleanup();
}

void* IOMQTT::run()
{
	while (running)
	{
		if (!mqttcpp->link)
		{
			connect();
		}
		usleep(100);
	}
	return 0;
}

void IOMQTT::connect()
{
	if (NULL == mqttcpp)
	{
		return;
	}
	logf = mqttcpp->link;
	int ret = mqttcpp->connect(ip_s.c_str(), port_i);
	if (MOSQ_ERR_SUCCESS != ret)
	{
		mqttcpp->link = false;
		if (logf!= mqttcpp->link) {
			CLogger::createInstance()->Log(MsgInfo
				, "mqttcpp connect(%s:%d) error!", ip_s.c_str(), port_i);
		}
		usleep(1000);
	}
	else {
		mqttcpp->link = true;
		if (logf != mqttcpp->link) {
			CLogger::createInstance()->Log(MsgInfo
				, "mqttcpp connect(%s:%d) success!", ip_s.c_str(), port_i);
		}
	}
}

void IOMQTT::disconnect()
{
	if (mqttcpp->link)
	{
		int ret = mqttcpp->disconnect();
		if (MOSQ_ERR_SUCCESS != ret)
		{
			Print_WARN("mqttcpp disconnect error\n");
		}
	}
}

void IOMQTT::cleanup()
{
	if (NULL != mqttcpp) {
		delete mqttcpp;
		mqttcpp = NULL;
	}
}
